#pragma once

#include "../base/list.h"
#include "lock.h"
#include <chrono>
#include <functional>
#include <thread>

OPEN_JLIB_NS


struct Executor {

private:
  typedef std::chrono::microseconds duration_t;
  typedef std::function<void()> mission_t;


  struct Mission {

    inline Mission (mission_t &&func, List *prev, List *next) noexcept
    : _func(std::move(func))
    , _list(prev, next) {
    }

    mission_t _func;
    List _list;
  };


  inline void _decIdleWorker () noexcept {
    std::lock_guard<SpinLock> lock{this->_status_lock};
    --this->_idle_worker_count;
  }

  inline void _incIdleWorker () noexcept {
    std::lock_guard<SpinLock> lock{this->_status_lock};
    ++this->_idle_worker_count;
  }


  struct Worker {

    inline Worker (Executor* executor, List *prev, List *next, bool core) noexcept
    : _thread(&Worker::run, this)
    , _executor(executor)
    , _list(prev, next)
    , _core(core) {
      _thread.detach();
    }

    void run () noexcept {
      _executor->_incIdleWorker();
      while (!_executor->_closing) {
        Mission* mission;
        if (_core) {
          mission = _executor->_takeMission();
          if (!mission)
            continue;
        }
        else {
          mission = _executor->_tryTakeMission();
          if (!mission)
            break;
        }
        _executor->_decIdleWorker();
        mission->_func();
        _executor->_incIdleWorker();
        delete mission;
      }
      _executor->_decIdleWorker();
      _executor->_removeWorker(this);
    }

    std::thread _thread;
    Executor* _executor;
    List _list;
    bool _core;
  };


  inline int _shouldNewWorker () noexcept {
    std::lock_guard<SpinLock> lock{_status_lock};
    if (_closing || _idle_worker_count > 0 || _pool_size >= _max_pool_size)
      return -1;
    ++_pool_size;
    return _pool_size <= _core_pool_size ? 1 : 0;
  }

  inline void _decWorkerCount () noexcept {
    std::lock_guard<SpinLock> lock{_status_lock};
    --_pool_size;
  }

  inline bool _shouldStartClose () noexcept {
    std::lock_guard<SpinLock> lock{_status_lock};
    if (_closed)
      return false;
    return _closing = true;
  }

  inline bool _checkAndClose () noexcept {
    std::lock_guard<SpinLock> lock{_status_lock};
    if (!_closed && _closing && !_pool_size)
      return _closed = true;
    return false;
  }

  bool _addWorker () noexcept {
    std::lock_guard<std::mutex> lock{_pool_mutex};
    auto hint = _shouldNewWorker();
    if (hint < 0) {
      _queue_condition.notify_one();
      return false;
    }
    new Worker(this, _pool.prev(), &_pool, (bool)hint);
    return true;
  }

  void _removeWorker (Worker* worker) noexcept {
    std::lock_guard<std::mutex> lock{_pool_mutex};
    worker->_list.detach();
    _decWorkerCount();
    delete worker;
    if (_checkAndClose())
      _pool_close_condition.notify_all();
  }


  void _addMission (std::function<void()> &&func) noexcept {
    std::unique_lock<std::mutex> lock{_queue_mutex};
    new Mission(std::move(func), _queue.prev(), &_queue);
  }

  Mission* _takeMission () noexcept {
    Mission* result = nullptr;
    std::unique_lock<std::mutex> lock{_queue_mutex};
    while (!_closing) {
      if (_queue.only()) {
        _queue_condition.wait(lock);
        continue;
      }
      auto list = _queue.next();
      list->detach();
      result = list->cast(&Mission::_list);
      break;
    }
    if (result)
      _addWorker();
    return result;
  }

  Mission* _tryTakeMission () noexcept {
    Mission* result = nullptr;
    std::unique_lock<std::mutex> lock{_queue_mutex};
    while (!_closing) {
      if (_queue.only()) {
        _queue_condition.wait_for(lock, _keep_alive_time);
        continue;
      }
      auto list = _queue.next();
      list->detach();
      result = list->cast(&Mission::_list);
      break;
    }
    if (result)
      _addWorker();
    return result;
  }


public:

  inline explicit Executor (size_t coreThreadSize = std::thread::hardware_concurrency(), size_t maxThreadSize = ~0UL, int64_t keepAliveTime = 60000000) noexcept
  : _keep_alive_time(keepAliveTime >= 0 ? keepAliveTime : 0) {
    _core_pool_size = coreThreadSize ? coreThreadSize : 1;
    _max_pool_size = maxThreadSize >= this->_core_pool_size ? maxThreadSize : this->_core_pool_size;
  }

  inline ~Executor () noexcept {
    shutdownNow();
  }


  inline void execute (mission_t &&func) noexcept {
    if (_closing)
      return;
    _addMission(std::forward<mission_t>(func));
    _addWorker();
  }


  inline void shutdown () noexcept {
    _closing = true;
  }

  void shutdownNow () noexcept {
    if (!_shouldStartClose())
      return;
    {
      std::unique_lock<std::mutex> lock {_queue_mutex};
      for (auto list = _queue.next(); list != &_queue; list = list->next())
        delete list->cast(&Mission::_list);
      _queue.clear();
      _queue_condition.notify_all();
    }
    {
      std::unique_lock<std::mutex> lock {_pool_mutex};
      while (!_closed)
        _pool_close_condition.wait(lock);
    }
  }


private:
  List _queue;
  std::condition_variable _queue_condition;
  std::mutex _queue_mutex;

  List _pool;
  std::condition_variable _pool_close_condition;
  std::mutex _pool_mutex;
  size_t _pool_size {0};
  size_t _core_pool_size;
  size_t _max_pool_size;

  SpinLock _status_lock;
  duration_t _keep_alive_time;
  size_t _idle_worker_count {0};
  volatile bool _closing {false};
  volatile bool _closed {false};
};


CLOSE_JLIB_NS